package olap4cloud.hadop.statistic;

import java.io.BufferedReader;
import java.io.DataInputStream;
import java.io.FileNotFoundException;
import java.io.IOException;
import java.io.InputStreamReader;
import java.sql.Date;
import java.text.SimpleDateFormat;
import java.util.concurrent.CompletionService;
import java.util.concurrent.ExecutorCompletionService;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;

import olap4cloud.hadop.statistic.StatisticsTable;

import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.FileStatus;
import org.apache.hadoop.fs.FileSystem;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.hbase.HBaseConfiguration;
import org.apache.hadoop.hbase.client.HTableInterface;
import org.apache.hadoop.hbase.client.HTablePool;
import org.apache.log4j.Logger;
import org.apache.log4j.xml.DOMConfigurator;

public class MainDriver {

	/**
	 * @uml.property  name="pool"
	 * @uml.associationEnd  multiplicity="(1 1)"
	 */
	private HTablePool pool;
	/**
	 * @uml.property  name="conf"
	 * @uml.associationEnd  multiplicity="(1 1)"
	 */
	private Configuration conf;

	private static int MAXPOOLHBASE = 20;
	private static int MAXTHREADPOOL = 500;

	private static Logger log1 = Logger.getLogger("logger");

	public MainDriver() {
		conf = HBaseConfiguration.create(new Configuration());
		pool = new HTablePool(conf, MAXPOOLHBASE);
	}

	/**
	 * Create hbase pool
	 * 
	 * @param pool
	 */
	public void initHBasePool(HTablePool pool) {

		HTableInterface[] tables = new HTableInterface[MAXPOOLHBASE];
		for (int n = 0; n < MAXPOOLHBASE; n++) {
			tables[n] = pool.getTable(StatisticsTable.getTableName());
		}

		for (int n = 0; n < MAXPOOLHBASE; n++) {
			try {
				pool.putTable(tables[n]);
			} catch (IOException e) {
				// TODO Auto-generated catch block
				e.printStackTrace();
			}
		}
	}

	/**
	 * Put result of job hadoop to hbase
	 * 
	 * @throws IOException
	 */
	public boolean increDataToHbase(String file) throws IOException {

		initHBasePool(pool);
		Configuration conf = new Configuration();

		conf.addResource("conf/core-site.xml");
		conf.addResource("conf/hbase-site.xml");
		
		System.out.println(conf.getRaw("fs.default.name"));
		System.out.println(conf.getRaw("hbase.rootdir"));

		FileSystem fs = FileSystem.get(conf);
		FileStatus[] fileStatus = fs.listStatus(new Path(file));

		long startTime = System.currentTimeMillis();
		for (FileStatus f : fileStatus) {
			if (f.getPath().toString().contains("part-r-")) {
				ExecutorService eservice = Executors
						.newFixedThreadPool(MAXTHREADPOOL);
				CompletionService<Object> objectCompletionService = new ExecutorCompletionService<Object>(
						eservice);
				IncreDataToHbase task = new IncreDataToHbase(pool, f.getPath());
				task.setEservice(eservice);
				task.setObjectCompletionService(objectCompletionService);
				try {
					task.call();
				} catch (Exception e) {
					// TODO Auto-generated catch block
					e.printStackTrace();
					return false;
				}
				eservice.shutdown();
				while (!eservice.isTerminated()) {
				}
				System.out.println("#");
			}
		}
		return true;
	}

	public static void main(String[] args) throws FileNotFoundException {
		DOMConfigurator.configure("log4j.xml");
		SimpleDateFormat simple = new SimpleDateFormat("yyyy-MM-dd:hh");
		System.out.println(simple.format(new Date(1365354000000l)));
		
		MainDriver main = new MainDriver();
		try {
			main.increDataToHbase(args[0]);
		} catch (IOException e) {
			// TODO Auto-generated catch block
			e.printStackTrace();
		}
	}
}
